package org.csdn.titan

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.graphx.{Edge, Graph, VertexId, VertexRDD}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}
import org.csdn.titan.utils.SparkUtils

object LogMaping {

  def main(args: Array[String]): Unit = {
    val spark:SparkSession = SparkUtils.getSparkSession(this.getClass.getName)

    val applog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-11/app")
    val weblog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-11/web")
    val wxapplog:Dataset[String] = spark.read.textFile("data/yiee_logs/2020-01-11/wxapp")



    val appIDS = extractIDs(applog)
    val webIDS = extractIDs(weblog)
    val wxIDS = extractIDs(wxapplog)

    val all_ids:RDD[Array[String]] = appIDS.union(webIDS).union(wxIDS)


    //构造点集合vertices
    val vertices:RDD[(Long,String)] = all_ids.flatMap(groups => {
      for (temp_id <- groups) yield (temp_id.hashCode.toLong, temp_id)
    })

    //构造边集合edges
    val edges:RDD[Edge[String]] = all_ids.flatMap(groups => {
      for (i <- 0 to groups.length - 2) yield Edge(groups(i).hashCode.toLong, groups(i + 1).hashCode.toLong, "")
    }).map(edge => (edge, 1))
      .reduceByKey(_ + _)
      .filter(tuple => tuple._2 > 2)
      .map(tuple => tuple._1)

    //构建图
    val graph = Graph(vertices, edges)
    //得到连通的点集
    val connectGraph = graph.connectedComponents()
    val result_vertices:VertexRDD[VertexId] = connectGraph.vertices
    //保存结果
    import spark.implicits._
    //result_vertices.toDF("id_hashcode","guid").write.csv("data/yiee_logs/2020-01-11/guid")
    result_vertices.toDF("id_hashcode","guid").write.parquet("data/yiee_logs/2020-01-11/guidmap")
    spark.close()
  }



  def extractIDs(logs:Dataset[String]):RDD[Array[String]] = {

    logs.rdd.map(line =>{
      val jsonObject = JSON.parseObject(line)
      val userObject = jsonObject.getJSONObject("user");
      val uid = userObject.getString("uid")
      val phoneObj = userObject.getJSONObject("phone")
      val imei = phoneObj.getString("imei")
      val mac = phoneObj.getString("mac")
      val imsi = phoneObj.getString("imsi")
      val androidId = phoneObj.getString("androidId")
      val deviceId = phoneObj.getString("deviceId")
      val uuid = phoneObj.getString("uuid")
      Array(uid,imei,mac,imsi,androidId,deviceId,uuid).filter(StringUtils.isNotBlank(_))
    })

  }



}
